package com.wk.net.heartbeat.handler;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import com.wk.net.heartbeat.HeartbeatTransferer;

/**
 * 主动对象：收集心跳信号并判断心跳源的状态
 * 
 * @author Bai Yu
 *
 * @param <HeartbeatType>
 *            心跳包类型
 */
public class HeartbeatHandler<HeartbeatType> {
	/**
	 * 创建心跳包接收器
	 * 
	 * @param transferer
	 *            心跳包的传输策略
	 * @param receiver
	 *            接收心跳包的底层实现
	 * @param checkPeriod
	 *            检查心跳的周期（单位：秒）
	 * @param absenceTolerance
	 *            掉线容忍度
	 */
	public HeartbeatHandler(HeartbeatTransferer<HeartbeatType> transferer, HeartbeatReceiver receiver, int checkPeriod,
			int absenceTolerance) {
		this.checkPeriod = checkPeriod;

		observer = new HeartbeatObserver(absenceTolerance);

		heartbeatListener = null;
		absenceListener = null;

		this.transferer = transferer;
		this.receiver = receiver;
	}

	public void observe(String name, boolean isOnline) {
		synchronized (observer) {
			observer.observe(name, isOnline);
		}
	}

	public void setHeartbeatListener(HeartbeatListener<HeartbeatType> listener) {
		this.heartbeatListener = listener;
	}

	public void setAbsenceListener(AbsenceListener listener) {
		this.absenceListener = listener;
	}

	public void start() {
		BlockingQueue<ParsedHeartbeatMessage<HeartbeatType>> parsedHeartbeats = null;

		if (heartbeatListener != null) {
			parsedHeartbeats = new LinkedBlockingQueue<>();
			new Thread(new HandleHeartbeat<>(parsedHeartbeats, heartbeatListener, observer)).start();
		}

		new Thread(new ReceiveHeartbeat<>(transferer, receiver, observer, parsedHeartbeats)).start();

		new ScheduledThreadPoolExecutor(1).scheduleAtFixedRate(new CheckHeartbeat(observer, absenceListener), 0,
				checkPeriod, TimeUnit.SECONDS);
	}

	private int checkPeriod;

	private HeartbeatObserver observer;

	private HeartbeatListener<HeartbeatType> heartbeatListener;
	private AbsenceListener absenceListener;

	private HeartbeatTransferer<HeartbeatType> transferer;
	private HeartbeatReceiver receiver;

}
